package com.yuma.tingshu.search.listener;

import com.rabbitmq.client.Channel;
import com.yuma.tingshu.album.client.AlbumInfoFeignClient;
import com.yuma.tingshu.album.client.CategoryFeignClient;
import com.yuma.tingshu.common.constant.RabbitConstant;
import com.yuma.tingshu.common.result.Result;
import com.yuma.tingshu.common.util.PinYinUtils;
import com.yuma.tingshu.model.album.AlbumAttributeValue;
import com.yuma.tingshu.model.album.BaseCategoryView;
import com.yuma.tingshu.model.search.AlbumInfoIndex;
import com.yuma.tingshu.model.search.AttributeValueIndex;
import com.yuma.tingshu.model.search.SuggestIndex;
import com.yuma.tingshu.model.user.UserInfo;
import com.yuma.tingshu.user.client.UserInfoFeignClient;
import com.yuma.tingshu.vo.album.AlbumListVo;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.BeanUtils;
import org.springframework.data.elasticsearch.client.elc.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;

@Slf4j
@Component
public class SearchListener {

    @Resource
    private UserInfoFeignClient userInfoFeignClient;

    @Resource
    private CategoryFeignClient categoryFeignClient;

    @Resource
    private AlbumInfoFeignClient albumInfoFeignClient;

    @Resource
    private ElasticsearchTemplate elasticsearchTemplate;

    /**
     * @RabbitListener：标记该方法为 RabbitMQ 消息监听器，当指定队列有消息时会触发方法执行
     * @QueueBinding：用于绑定队列、交换机和路由键的关系，相当于在代码中声明绑定关系
     * @Queue：指定要监听的队列名称（从常量类获取）
     * @Exchange：指定交换机信息，包括名称、类型（这里是 TOPIC 类型）和是否忽略声明异常
     * key：指定路由键，只有匹配该路由键的消息才会被当前方法处理
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(RabbitConstant.ALBUM_SAVE_QUEUE),
            exchange = @Exchange(value = RabbitConstant.ALBUM_EXCHANGE, type = ExchangeTypes.TOPIC, ignoreDeclarationExceptions = "true"),
            key = {RabbitConstant.ALBUM_SAVE_RK}
    ))
    public void saveAlbum(Long albumId, Channel channel, Message message) throws IOException {
        try {
            if (albumId == null) {
                // 游标， 不要批量确认消息，可能导致消息丢失
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                return;
            }
            Result<AlbumListVo> albumListVoResult = this.albumInfoFeignClient.getAlbumListVoById(albumId);
            Assert.notNull(albumListVoResult, "实时同步专辑数据时，获取专辑信息失败");
            AlbumListVo albumListVo = albumListVoResult.getData();
            Assert.notNull(albumListVo, "实时同步专辑数据时，专辑信息不存在" + albumId);

            // 执行业务操作，保存到索引库
            AlbumInfoIndex infoIndex = new AlbumInfoIndex();
            BeanUtils.copyProperties(albumListVo, infoIndex);
            infoIndex.setId(albumListVo.getAlbumId());
            // 根据主播ID查询
            Result<UserInfo> userInfoResult = userInfoFeignClient.getUserById(albumListVo.getUserId());
            Assert.notNull(userInfoResult, "数据导入时，获取用户信息失败");
            UserInfo userInfo = userInfoResult.getData();
            if (userInfo != null) {
                infoIndex.setAnnouncerName(userInfo.getNickname());
            }
            // 根据三级分类id查询一二三级分类
            Result<BaseCategoryView> baseCategoryViewResult = categoryFeignClient.getBaseCategoryViewByCategory3Id(albumListVo.getCategory3Id());
            Assert.notNull(baseCategoryViewResult, "数据导入时，获取三级分类信息失败");
            BaseCategoryView baseCategoryView = baseCategoryViewResult.getData();
            if (baseCategoryView != null) {
                infoIndex.setCategory1Id(baseCategoryView.getCategory1Id());
                infoIndex.setCategory2Id(baseCategoryView.getCategory2Id());
            }
            // 伪造数据
            infoIndex.setPlayStatNum(ThreadLocalRandom.current().nextInt(10000) + 10000);
            infoIndex.setSubscribeStatNum(ThreadLocalRandom.current().nextInt(10000) + 10000);
            infoIndex.setBuyStatNum(ThreadLocalRandom.current().nextInt(10000) + 10000);
            infoIndex.setCommentStatNum(ThreadLocalRandom.current().nextInt(10000) + 10000);
            infoIndex.setHotScore(infoIndex.getPlayStatNum() * 0.2 + infoIndex.getCommentStatNum() * 0.2
                    + infoIndex.getBuyStatNum() * 0.3 + infoIndex.getSubscribeStatNum() * 0.3);
            // 属性信息
            Result<List<AlbumAttributeValue>> albumAttributeValuesResult = albumInfoFeignClient.findAlbumAttributeValuesByAlbumId(albumListVo.getAlbumId());
            Assert.notNull(albumAttributeValuesResult, "数据导入时，获取专辑标签值失败！");
            List<AlbumAttributeValue> albumAttributeValues = albumAttributeValuesResult.getData();
            if (!CollectionUtils.isEmpty(albumAttributeValues)) {
                // 把AlbumAttributeValue集合转化成AttributeValueIndex集合
                List<AttributeValueIndex> attributeValueIndexList = albumAttributeValues.stream().map(albumAttributeValue -> {
                    AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                    BeanUtils.copyProperties(albumAttributeValue, attributeValueIndex);
                    return attributeValueIndex;
                }).collect(Collectors.toList());
                infoIndex.setAttributeValueIndexList(attributeValueIndexList);
            }

            this.elasticsearchTemplate.save(infoIndex);

            /// 保存数据到提示词库
            // 标题
            SuggestIndex titleSuggestIndex = new SuggestIndex();
            titleSuggestIndex.setId("title_" + albumListVo.getAlbumId()); // 需要指定id，以方便将来实时同步数据。
            titleSuggestIndex.setTitle(albumListVo.getAlbumTitle());
            titleSuggestIndex.setKeyword(new Completion(List.of(albumListVo.getAlbumTitle())));
            titleSuggestIndex.setKeywordPinyin(new Completion(List.of(PinYinUtils.toHanyuPinyin(albumListVo.getAlbumTitle()))));
            titleSuggestIndex.setKeywordSequence(new Completion(List.of(PinYinUtils.getFirstLetter(albumListVo.getAlbumTitle()))));
            elasticsearchTemplate.save(titleSuggestIndex);
            // 简介
            SuggestIndex introSuggestIndex = new SuggestIndex();
            introSuggestIndex.setId("intro_" + albumListVo.getAlbumId()); // 需要指定id，以方便将来实时同步数据。
            introSuggestIndex.setTitle(albumListVo.getAlbumIntro());
            introSuggestIndex.setKeyword(new Completion(List.of(albumListVo.getAlbumIntro())));
            introSuggestIndex.setKeywordPinyin(new Completion(List.of(PinYinUtils.toHanyuPinyin(albumListVo.getAlbumIntro()))));
            introSuggestIndex.setKeywordSequence(new Completion(List.of(PinYinUtils.getFirstLetter(albumListVo.getAlbumIntro()))));
            elasticsearchTemplate.save(introSuggestIndex);
            // 主播
            if (userInfo != null){
                SuggestIndex announceSuggestIndex = new SuggestIndex();
                announceSuggestIndex.setId("announce_" + userInfo.getId()); // 需要指定id，以方便将来实时同步数据。
                announceSuggestIndex.setTitle(userInfo.getNickname());
                announceSuggestIndex.setKeyword(new Completion(List.of(userInfo.getNickname())));
                announceSuggestIndex.setKeywordPinyin(new Completion(List.of(PinYinUtils.toHanyuPinyin(userInfo.getNickname()))));
                announceSuggestIndex.setKeywordSequence(new Completion(List.of(PinYinUtils.getFirstLetter(userInfo.getNickname()))));
                elasticsearchTemplate.save(announceSuggestIndex);
            }

            // 手动确认消息
            /**
             * 第一个参数 DeliveryTag：消息的唯一标识，每个消息在信道（Channel）中都有一个递增的标签
             * 第二个参数 multiple：是否批量确认。false 表示只确认当前 DeliveryTag 对应的消息；
             * true 表示确认该标签及之前所有未确认的消息。这里用 false 是精准确认当前消息，更安全。
             */
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            if (message.getMessageProperties().getRedelivered()/*消息的 Redelivered 属性为 true，表示该消息之前已被投递过一次，但因未被确认而重新入队，现在是第二次（或多次）投递*/) {
                // 如果消息重复投递过，则不重新入队，该消息变成死信消息，如果对应的队列绑定了死信队列则进入死信队列，否则该消息会丢失
                // channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                /**
                 * 第一个参数：消息的 DeliveryTag
                 * 第二个参数 requeue：是否重新入队。false 表示不重新入队
                 */
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                // 记录日志
                log.error("ES同步数据过程中，消费消费失败！消息的内容：{}", albumId);
            } else {
                // 否则重新入队进行重试
                /**
                 * 第一个参数：消息的 DeliveryTag
                 * 第二个参数 multiple：是否批量拒绝，false 表示只拒绝当前消息
                 * 第三个参数 requeue：是否重新入队，true 表示将消息重新放入队列，等待再次投递
                 */
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            }
        }
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(RabbitConstant.ALBUM_DELETE_QUEUE),
            exchange = @Exchange(value = RabbitConstant.ALBUM_EXCHANGE, type = ExchangeTypes.TOPIC, ignoreDeclarationExceptions = "true"),
            key = {RabbitConstant.ALBUM_DELETE_RK}
    ))
    public void delAlbum(Long albumId, Channel channel, Message message) throws IOException {
        try {
            if (albumId == null) {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                return;
            }
            // 删除对应索引库数据
            this.elasticsearchTemplate.delete(albumId.toString(), AlbumInfoIndex.class);

            // 删除提示词库数据
            //this.elasticsearchTemplate.delete(albumId.toString(), SuggestIndex.class); X,因为提示词索引中ID不为albumId
            this.elasticsearchTemplate.delete("title_"+ albumId, SuggestIndex.class);
            this.elasticsearchTemplate.delete("intro_"+ albumId, SuggestIndex.class);
            this.elasticsearchTemplate.delete("announce_"+ albumId, SuggestIndex.class);


            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            if (message.getMessageProperties().getRedelivered()) {
                // 如果消息重复投递过，则不重新入队，该消息变成死信消息，如果对应的队列绑定了死信队列则进入死信队列，否则该消息会丢失
                // channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                // 记录日志
                log.error("ES同步数据过程中，消费消费失败！消息的内容：{}", albumId);
            } else {
                // 否则重新入队进行重试
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            }
        }
    }
}